/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.pig.pigpen.view;

import java.util.Iterator;

import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.LogicalOperator;
import org.apache.pig.pigpen.ConsoleLogger;

//We consider an Identifier a pig variable. In the context of 
//Example Generator each variable will be associated with a databag
//containing example data
public class PigVariable implements IDataContainer {

    private LogicalOperator op;
    private DataContainer data;
    private Object parent;

    ConsoleLogger logger = ConsoleLogger.getDefault();

    public PigVariable(LogicalOperator op, DataBag data, ExampleData parent) {
        this.op = op;
        try {
            this.data = new DataContainer(data, parent, op.getSchema());
        } catch (FrontendException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            logger.error("Error obtaining schema\n" + e.getMessage());
        }
        this.parent = parent;
    }

    public String getName() {
        try {
            return op.getAlias() + " " + op.getSchema();
        } catch (FrontendException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            logger.error("Error obtaining schema\n" + e.getMessage());
        }
        return null;
    }

    public String getShortName() {
        return op.getAlias() + " " + data.getData().toString();
    }

    public String getType() {
        return data.getType();
    }

    public DataContainer getData() {
        return data;
    }

    public Object getParent() {
        return parent;
    }

    @Override
    public String toString() {
        return op.getAlias() + " : " + data.toString();
    }

    public Object[] getChildren() {
        IDataContainer[] result = new IDataContainer[data.getChildren().length];
        DataBag bag = (DataBag) data.getData();
        int i = 0;
        for (Iterator<Tuple> it = bag.iterator(); it.hasNext();) {
            try {
                result[i++] = new DataContainer(it.next(), data.getData(), op
                        .getSchema());

            } catch (FrontendException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

        return result;
    }

}
